﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.IO;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;

namespace System.Threading.Actors
{
    // A wrapper around a TcpClient that allows a client to speak "actor-ese" to an actor over the raw TcpClient
    public class ActorRawTcpClient : IDisposable
    {
#pragma warning disable 3026  // protected volatile fields aren't CLS compliant
        protected volatile bool m_disposed = false;
#pragma warning restore 3026

        private volatile bool m_faulted = false;
        private volatile bool m_shuttingDown;

        private ITcpClient m_tcpClient; // passed in constructor
        // In an ideal world we could use simply one bidirectional stream, such as what the .NET Net Classes will return.
        // However, WinRT will return two streams for a socket - an input stream and an output stream.
        private Stream m_inputStream;
        private Stream m_outputStream;

        private AsyncQueue<ActorEvent> m_eventQueue;
        private readonly IObserver<ActorEvent> m_eventObserver;
        private readonly Dictionary<int, TaskCompletionSource<ActorServiceResult>> m_waiterMap =
            new Dictionary<int, TaskCompletionSource<ActorServiceResult>>();
        private readonly SemaphoreSlim m_waiterMapSem = new SemaphoreSlim(1, 1);

        /// <summary>
        /// Constructs an ActorRawTcpClient.
        /// </summary>
        public ActorRawTcpClient(ITcpClient tcpClient, IObserver<ActorEvent> eventObserver, string uniquePathInfo, string partitionName)
        {
            Contract.Requires(tcpClient != null);

            m_tcpClient = tcpClient;
            m_inputStream = tcpClient.InputStream;
            m_outputStream = tcpClient.OutputStream;
            m_eventObserver = eventObserver;
            SetupConnectionAsync(uniquePathInfo, partitionName).Wait();
        }

        private ActorRawTcpClient(ITcpClient tcpClient, IObserver<ActorEvent> eventObserver, string uniquePathInfo, string partitionName, bool bogus)
        {
            m_tcpClient = tcpClient;
            m_inputStream = tcpClient.InputStream;
            m_outputStream = tcpClient.OutputStream;
            m_eventObserver = eventObserver;
            // DON'T do SetupConnection, yet.
        }

        public static async Task<ActorRawTcpClient> CreateAsync(ITcpClient tcpClient, IObserver<ActorEvent> eventObserver, string uniquePathInto, string partitionName)
        {
            ActorRawTcpClient result = new ActorRawTcpClient(tcpClient, eventObserver, uniquePathInto, partitionName, bogus: false);
            await result.SetupConnectionAsync(uniquePathInto, partitionName).ConfigureAwait(continueOnCapturedContext:false);
            return result;
        }

        // Effectively closes the client
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            m_shuttingDown = true;
            if (!m_disposed)
            {
                if (m_tcpClient != null)
                {
                    m_tcpClient.Dispose(); // Takes care of disposing NetworkStream as well
                    m_tcpClient = null;
                }
                if (m_eventQueue != null)
                {
                    m_eventQueue.CompleteAdding(true);
                    m_eventQueue = null;
                }

                FaultAllWaiters();
                m_disposed = true;
            }
        }

        public event EventHandler Faulted;

        public bool IsFaulted
        {
            get { return m_faulted; }
            protected set { m_faulted = value; }
        }

        public bool IsShuttingDown {
            get { return m_shuttingDown; }
        }

        // Called when we are being closed or something catastrophic has happened.  
        // Faults all waiting tasks.
        private void FaultAllWaiters()
        {
            try
            {
                m_waiterMapSem.Wait();
                foreach (var waiter in m_waiterMap.Values)
                {
                    waiter.TrySetException(new OperationCanceledException("Disconnected from actor"));
                }
                m_waiterMap.Clear();
            }
            finally { m_waiterMapSem.Release(); }
        }

        // Mark ourself as faulted, notifying any interested subscribers to the Faulted event
        protected void MarkFaulted()
        {
            if (!IsFaulted)
            {
                IsFaulted = true;
                var faulted = Faulted;
                if (faulted != null && !m_disposed)
                    faulted(this, new EventArgs());
                FaultAllWaiters(); // Recover first, then let waiters know that they have faulted
            }
        }

        protected async Task SetupConnectionAsync(string uniquePathInfo, string partitionName)
        {
            // Connect to a specific back-end target, if appropriate.
            // For connecting to the Azure gateway, the target will be an actor name, and if necessary, a # then a partition name
            // For connecting to the ActorRawTcpListener, the target will be the unique "path" portion of the service's URI.
            // In all cases, "OK" should be returned.
            string pathAndPartitionInfo = uniquePathInfo;
            if (partitionName != null)
                pathAndPartitionInfo += "#" + partitionName;
            Utilities.NetworkHelpers.WriteString(pathAndPartitionInfo, m_outputStream);
            var buffer = new byte[128];
            var status = await Utilities.NetworkHelpers.ReadStringAsync(m_inputStream, buffer).ConfigureAwait(continueOnCapturedContext:false);

            if (!status.Equals("OK"))
            {
                Dispose();
                if (status.StartsWith("EXCEPTION:"))
                {
                    throw CreateException(uniquePathInfo, status);
                }
                else
                    throw new InvalidOperationException("Could not connect with uniquePathInfo " + uniquePathInfo + ", status = " + status);
            }

            // We need to escape any lock that we may be holding for this method call, as ReaderLoop() has
            // some potential awaits in it.  So fire it off in another task.  However, we aren't using locks anymore.
            // Try running this inline for now instead of using Task.Run(() => ReaderLoop());
            ReaderLoop();  // Fire and forget
        }

        protected async void ActorEventQueueRunLoop()
        {
            while (m_eventQueue != null)  // while not disposed...
            {
                try
                {
                    ActorEvent aEvent = await m_eventQueue.DequeueAsync().ConfigureAwait(continueOnCapturedContext: false);
                    m_eventObserver.OnNext(aEvent);
                }
                catch (EndOfQueueException)
                {
                    break;
                }
                catch (InvalidOperationException)  // Note: ObjectDisposedException is a subclass of InvalidOperationException
                {
                    break;
                }
                catch (Exception e)
                {
                    Debug.Assert(false, "Got an unexpected exception in ActorEventQueueRunLoop!  " + e);
                    throw;
                }
            }
        }

        protected static Exception CreateException(string uniquePathInfo, string status)
        {
            Contract.Requires(status != null);
            Contract.Requires(status.StartsWith("EXCEPTION:"));

            int secondColon = status.IndexOf(':', 11);
            String exceptionType = status.Substring(10, secondColon - 10);
            String exceptionMessage = status.Substring(secondColon + 1);
            Exception ex = ActorClient.ConstructException(exceptionType, exceptionMessage);
            ActorNotFoundException anf = ex as ActorNotFoundException;
            if (anf != null)
                anf.UniquePathInfo = uniquePathInfo;
            else
                ex.Data["UniquePathInfo"] = uniquePathInfo;
            return ex;
        }

        // Loop on feedback from the actor
        protected async void ReaderLoop()
        {
            var shortMessageBuffer = new byte[1024];
            try
            {
                while (!m_disposed && !m_shuttingDown && m_inputStream.CanRead)
                {
                    ActorServiceResult asr = null;
                    ActorEvent ae = null;
                    var result = await SerializationHelper.DeserializeActorFeedbackAsync(
                        m_inputStream,
                        shortMessageBuffer).ConfigureAwait(continueOnCapturedContext: false);
                    if ((ae = result as ActorEvent) != null)
                    {
                        // For events, we need to pass them through the registered event observer.
                        // However, the observer may call actor methods.  Run the event processing on another thread
                        // to avoid reentrancy here (otherwise this thread in ReaderLoop could deadlock with itself).
                        if (m_eventQueue == null)
                        {
                            m_eventQueue = new AsyncQueue<ActorEvent>();
                            ActorEventQueueRunLoop();
                        }
                        m_eventQueue.Enqueue(ae, true);
                    }
                    else if ((asr = result as ActorServiceResult) != null)
                    {
                        // For request responses, complete the associated waiting task
                        var seqNo = asr.ClientSequenceNumber;
                        TaskCompletionSource<ActorServiceResult> tcs = null;
                        try
                        {
                            await m_waiterMapSem.WaitAsync().ConfigureAwait(continueOnCapturedContext:false);
                            if (m_waiterMap.TryGetValue(seqNo, out tcs))
                                m_waiterMap.Remove(seqNo);
                        }
                        finally { m_waiterMapSem.Release(); }

                        if (tcs != null)
                            tcs.TrySetResult(asr);  // Runs continuations.  Consider doing this on another thread?
                        else
                        {
                            if (!m_disposed)
                            {
                                Trace.WriteLine("ActorRawTcpClient::ReaderLoop: Got an ActorServiceResult back, but no one was waiting for it!  We either faulted this request, or message sent to wrong client?  Client Sequence Number: {0}", asr.ClientSequenceNumber);
                            }
                        }
                    }
                    else
                        Contract.Assume(false, "Unrecognized deserialized value in ActorRawTcpClient::ReaderLoop.  Got: " + result);
                }
            }
            catch (EndOfStreamException)
            {
                // This is OK -- The client on the other side is closed.
                m_inputStream.Dispose();
                if (!m_shuttingDown)
                {
                    // This *may* be due to the target actor going down.  So mark ourselves as faulted so that we can retry.
                    MarkFaulted();
                }
            }
            catch (IOException)
            {
                MarkFaulted();
            }
            catch (ObjectDisposedException)
            {
                // This is OK -- someone disposed of us
            }
            catch (Exception e)
            {
                Console.WriteLine("ActorRawTcpClient.ReaderLoop went down unpredictably: {0}", e);
            }
            finally
            {
                FaultAllWaiters();
            }
        }

        // Transmit an ActorServiceRequest to the actor
        public async Task<ActorServiceResult> TransmitRequestAsync(ActorServiceRequest request)
        {
            Contract.Requires(request != null);

            if (m_outputStream == null)
                throw new ObjectDisposedException("ActorRawTcpClient");

            // This causes some problems during teardown, as we can catch-and-swallow exceptions
            // but not assertions.  Let's change it to an exception.
            //Contract.Assert(m_outputStream.CanWrite);
            if (!m_outputStream.CanWrite)
                throw new InvalidOperationException("Attempted transmission on unwriteable stream");


            if (request.ClientSequenceNumber == ActorServiceRequest.FireAndForgetSequenceNumber)
            {
                // This is a "fire-and-forget" request
                SerializationHelper.SerializeActorServiceRequest(request, m_outputStream);
                return null;
            }

            // Create a holding place for the eventual result
            int clientSequenceNumber = request.ClientSequenceNumber;
            var tcs = new TaskCompletionSource<ActorServiceResult>();
            try
            {
                await m_waiterMapSem.WaitAsync().ConfigureAwait(continueOnCapturedContext:false);
				if (clientSequenceNumber != ActorServiceRequest.FireAndForgetSequenceNumber)
                    m_waiterMap[clientSequenceNumber] = tcs;
            }
            finally { m_waiterMapSem.Release(); }


            try
            {
                // Shoots request to the actor
                SerializationHelper.SerializeActorServiceRequest(request, m_outputStream);
            }
            catch (Exception e)
            {
                if (e is EndOfStreamException)
                {
                    m_outputStream.Dispose();
                    // End-of-stream may be due to actor going down.  So mark ourselves as faulted to cause a retry
                    if (!m_shuttingDown)
                        MarkFaulted();
                }
                else if (e is IOException)
                {
                    if (!m_shuttingDown)
                        MarkFaulted();
                }
                else
                {
                    tcs.TrySetException(e); // Make sure that the request gets completed regardless...
                    //if (!(e is ObjectDisposedException))
                    //    Console.WriteLine("Unexpected exception from ARTC.TransmitRequest: {0}", e.Message);
                }
            }

            // @TODO: Should we be catching any exceptions here?
            return await tcs.Task.ConfigureAwait(continueOnCapturedContext:false);
        }
    }
}
